package com.hexiang.rocketmq.producer;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.acl.common.AclClientRPCHook;
import org.apache.rocketmq.acl.common.SessionCredentials;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.RPCHook;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.nio.charset.StandardCharsets;

@Component
@Slf4j
public class TraceMQProducer {

    private final String TOPIC = "test_trace_topic";

    private DefaultMQProducer defaultMQProducer;

    private String producerGroup="test_trace_producer_group";

    @Value("${rocketmq.name-server}")
    private String nameServer;

    @Value("${rocketmq.producer.send-message-timeout}")
    private Integer messageTimeOut;


    @PostConstruct
    public void start() throws MQClientException {
        defaultMQProducer = new DefaultMQProducer(producerGroup,true);
        defaultMQProducer.setNamesrvAddr(nameServer);
        defaultMQProducer.start();
    }

    @PreDestroy
    public void shutdown() {
        defaultMQProducer.shutdown();
    }

    RPCHook getAclRPCHook() {
        return new AclClientRPCHook(new SessionCredentials("rocketmq_u1", "12345678"));
    }

    /**
     * 发送普通消息
     */
    public void sendMsg(String msgBody) throws InterruptedException, RemotingException, MQClientException, MQBrokerException {
        Message message = new Message(TOPIC, (msgBody).getBytes(StandardCharsets.UTF_8));
        SendResult sendResult = defaultMQProducer.send(message);
        log.warn("sendResult " + sendResult);
    }

}